package com.akkainaction.cluster.addcluster;

import akka.actor.ActorSystem;
import akka.actor.Address;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.cluster.Cluster;
import akka.cluster.Member;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

import static akka.cluster.ClusterEvent.*;

public class Node04 extends UntypedActor {

    Cluster cluster = Cluster.get(getContext().system());


    @Override
    public void preStart() {
        cluster.subscribe(getSelf(), initialStateAsEvents(), UnreachableMember.class, MemberEvent.class);
    }

    @Override
    public void postStop() throws Exception {
        cluster.unsubscribe(getSelf());
    }

    @Override
    public void onReceive(Object msg) throws Exception {
        if (msg instanceof MemberUp) {
            MemberUp memberUp = (MemberUp) msg;
            Member m = memberUp.member();
            System.out.println(getSelf() + "-->事件:" + memberUp + "-->role:" + m.roles());
        } else if (msg instanceof UnreachableMember) {
            UnreachableMember unreachableMember = (UnreachableMember) msg;
            Member m = unreachableMember.member();
            System.out.println(getSelf() + "-->事件:" + unreachableMember + "-->role:" + m.roles());
        } else if (msg instanceof MemberRemoved) {
            MemberRemoved memberRemoved = (MemberRemoved) msg;
            Member m = memberRemoved.member();
            System.out.println(getSelf() + "-->事件:" + memberRemoved + "-->role:" + m.roles());
        } else if (msg instanceof MemberEvent) {
            MemberEvent me = (MemberEvent) msg;
            Member m = me.member();
            System.out.println(getSelf() + "-->事件:" + me + "-->role:" + m.roles());
        } else {
            System.out.println(getSelf() + "-->Other: " + msg);
            unhandled(msg);
        }
    }

    public static void main(String[] args) {
        Config config = ConfigFactory.load("akka-cluster-node04.conf");
        ActorSystem system = ActorSystem.create("sys", config);
        system.actorOf(Props.create(Node04.class), "node04");
        Cluster cluster = Cluster.get(system);
        Address address = new Address("akka.tcp", "sys", "127.0.0.1", 2551);
        cluster.join(address);

        CurrentClusterState state= Cluster.get(system).state();
        System.out.println("Leader节点:"+state.getLeader());
        System.out.println("所有节点列表:"+state.getMembers());
        System.out.println("unreachable节点列表:"+state.getUnreachable());
    }
}
